package sc.writable;


import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;

import org.apache.hadoop.io.VIntWritable;
import org.apache.hadoop.io.WritableComparable;
import org.apache.hadoop.io.WritableComparator;
import org.apache.hadoop.io.WritableUtils;


/**
 * Paire de int qui peut être utilisée comme clé pour du MapReduce.
 * 
 * Les données sont écrites en entier variable et sont donc stockées sur moins
 * de 32 bits si cela est possible.
 * 
 * @author Quentin Rossé
 *
 */
public class IntegerPair implements WritableComparable<IntegerPair>{

	private int i1;
	private int i2;
	
	private static VIntWritable vint = new VIntWritable();
	
	//for compatibility (Writable)
	public IntegerPair(){}
	
	public IntegerPair(int i1, int i2){
		this.i1 = i1;
		this.i2 = i2;
	}
	
	public int getI1() {
		return i1;
	}

	public int getI2() {
		return i2;
	}

	public void setPair(int i1, int i2) {
		this.i1 = i1;
		this.i2 = i2;
	}

	@Override
	public void readFields(DataInput in) throws IOException {
		vint.readFields(in);
		i1 = vint.get();
		vint.readFields(in);
		i2 = vint.get();
	}

	@Override
	public void write(DataOutput out) throws IOException {
		vint.set(i1);
		vint.write(out);
		vint.set(i2);
		vint.write(out);
	}

	/**
	 * Si les deux i1 sont différents on retroune leur comparaison
	 * s'ils sont identiques on compare les deux i2
	 */
	@Override
	public int compareTo(IntegerPair pair) {
		int comp = i1 - pair.i1;
		if(comp == 0)
			return i2 - pair.i2;
		return comp;
	}
	
	public static class Comparator extends WritableComparator {
		public Comparator() {
			super(IntegerPair.class);
	    }
		
		
		@Override
	    public int compare(byte[] b1, int s1, int l1,  //val1
	                       byte[] b2, int s2, int l2){ //val2
	    	int tmp;
			try{
		    	//compare les premiers nombres de la paire
				int v1 = readVInt(b1, s1);
		    	int v2 = readVInt(b2, s2);
		    	tmp = v1 - v2;
		    	if(tmp != 0)
		    		return tmp;
		    	
		    	s1 += WritableUtils.decodeVIntSize(b1[s1]);
		    	s2 += WritableUtils.decodeVIntSize(b2[s2]);
		    	//compare les deuxièmes nombres de la paire
		    	v1 = readVInt(b1, s1);
		    	v2 = readVInt(b2, s2);
		    	return v1 - v2;
		    	
	    	}catch(IOException e){
	    		e.printStackTrace();
	    		throw new RuntimeException("Error while comparing " + 
	    				IntegerPair.class.getName());
	    	}
	    }
	}

	//définition du comparateur optimisé pour que Hadoop l'utilise 
	//automatiquement à la place de l'autre
	static {
		WritableComparator.define(IntegerPair.class, new Comparator());
	}

}
